import base64
import asyncio
import logging
from datetime import datetime
from typing import Optional

from pymongo import DESCENDING
from pytz import timezone

from client.openai_client import invoke_laozhangapi_only_result
from model.wristband_models import WarningData, AudioData
from utils.aliyun_token_util import send_asr
from utils.mongo_util import db

logger = logging.getLogger(__name__)


def b64_to_amr_bytes(voice_b64: str) -> bytes:
    """把 JSON 里的 voice 字段转成 AMR 原始字节"""
    return base64.b64decode(voice_b64)


async def handle_audio(data: AudioData) -> Optional[str]:
    """优化的音频处理函数，使用异步操作提高性能"""
    try:
        # 步骤1: 异步执行语音识别
        asr_text = await _async_send_asr(b64_to_amr_bytes(data.audio.voice))
        if not asr_text:
            logger.warning(f"ASR识别失败，IMEI: {data.device_id}")
            return None
        
        # 步骤2: 异步AI意图识别
        prompt = (
            f'我下面的话要判断他的意思，你就输出一个单词，'
            f'如果是表示收到了就输出2，表示完成了就输出3，'
            f'如果表示拒绝就输出0，没有以上三种含义就输出-1'
            f' 不用多说任何一个字。 {asr_text}'
        )
        answer = await invoke_laozhangapi_only_result(prompt)
        
        # 步骤3: 处理响应结果
        if answer != '-1':
            await _update_order_status(data.device_id, answer, asr_text, data.timestamp)
            logger.info(f"音频处理完成，IMEI: {data.device_id}, 结果: {answer}")
        
        return answer
        
    except Exception as e:
        logger.error(f"音频处理失败，IMEI: {data.device_id}, 错误: {e}")
        return None


async def _async_send_asr(audio_bytes: bytes) -> Optional[str]:
    """异步语音识别函数"""
    try:
        # 使用线程池执行阻塞的ASR调用
        asr_result = await asyncio.to_thread(send_asr, audio_bytes)
        return asr_result
    except Exception as e:
        logger.error(f"ASR调用失败: {e}")
        return None


async def _update_order_status(device_id: str, answer: str, message: str, timestamp: int):
    """异步更新订单状态"""
    try:
        date = datetime.fromtimestamp(timestamp, timezone('Asia/Shanghai'))
        
        if answer == '2':  # 收到
            await _async_mongo_update(
                {"handler.imei": device_id, 'state': int(answer)},
                {"$set": {
                    'state': 3,
                    'handlingTime': date,
                    'handlingMessage': message
                }}
            )
        elif answer == '3':  # 完成
            await _async_mongo_update(
                {"handler.imei": device_id, 'state': int(answer)},
                {"$set": {
                    'state': 4,
                    'completionTime': date,
                    'completionMessage': message
                }}
            )
    except Exception as e:
        logger.error(f"更新订单状态失败: {e}")


async def _async_mongo_update(filter_doc: dict, update_doc: dict):
    """异步MongoDB更新操作包装器"""
    try:
        # 使用线程池执行同步MongoDB操作
        await asyncio.to_thread(
            db.s_clean_keeping_order.find_one_and_update,
            filter_doc,
            update_doc,
            sort=[("createDate", DESCENDING)]
        )
    except Exception as e:
        logger.error(f"MongoDB更新失败: {e}")
        raise
